package march.finalmarch21.march01.plan1;

import march.finalmarch21.FlowBean;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.conf.Configured;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileSplit;
import org.apache.hadoop.mapreduce.lib.input.TextInputFormat;
import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat;
import org.apache.hadoop.util.Tool;
import org.apache.hadoop.util.ToolRunner;

import java.io.IOException;
import java.math.BigDecimal;
import java.util.HashMap;
import java.util.Map;

public class March01 extends Configured implements Tool {
    public static Map<String, BigDecimal> huiLvMap = new HashMap<>();
    public static Map<String,String> huiLvNameMap = new HashMap<>();

    static class March01Mapper extends Mapper<LongWritable, Text,Text,NullWritable>{
        String fileName;

        @Override
        protected void setup(Context context) throws IOException, InterruptedException {
            FileSplit inputSplit = (FileSplit) context.getInputSplit();
            fileName = inputSplit.getPath().getName();
        }
        @Override
        protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
            String[] split = value.toString().split(",");

            if(fileName.contains("huilv")){
                //汇率表，使用map进行存储，方便Reduce阶段取
                huiLvMap.put(split[0],new BigDecimal(split[1]));
                huiLvNameMap.put(split[0],split[2]);
            }else{
                context.write(value,NullWritable.get());
            }
        }
    }

    static class March01Reducer extends Reducer<Text,NullWritable, NullWritable,FlowBean>{

        @Override
        protected void reduce(Text key, Iterable<NullWritable> values, Context context) throws IOException, InterruptedException {
            String[] split = key.toString().split(",");
            String id = split[0];
            String industry = split[1];
            String type = split[2];
            String coinId = split[3];
            String loanAmount = split[4];
            String drawMoneyAmount = split[5];
            String repaidAmount = split[6];
            String interestAmount = split[7];

            FlowBean flowBean = new FlowBean();
            //数据填充
            flowBean.setId(id);
            flowBean.setIndustry(industry);
            flowBean.setType(type);
            flowBean.setCoinId(coinId);
            flowBean.setLoanAmount(loanAmount);
            flowBean.setDrawMoneyAmount(drawMoneyAmount);
            flowBean.setRepaidAmount(repaidAmount);
            flowBean.setInterestAmount(interestAmount);

            BigDecimal huiLv = huiLvMap.get(coinId);
            String coinName = huiLvNameMap.get(coinId);

            flowBean.setCoinId(coinId);
            flowBean.setCoinName(coinName);
            flowBean.setExchangePrice(huiLv.toString());

            //考虑到汇率计算，所以在这里使用BigDecimal，而不是使用浮点数型
            BigDecimal loanRMB = new BigDecimal(loanAmount).multiply(huiLv);
            BigDecimal drawRMB = new BigDecimal(drawMoneyAmount).multiply(huiLv);
            BigDecimal repaidRMB = new BigDecimal(repaidAmount).multiply(huiLv);
            BigDecimal interestRMB = new BigDecimal(interestAmount).multiply(huiLv);
            //贷款余额=实际提款额-已还本金额
            BigDecimal balanceRMB = drawRMB.subtract(repaidRMB);

            //数据填充
            flowBean.setCoinName(coinName);
            flowBean.setLoanRMB(loanRMB.toString());
            flowBean.setDrawRMB(drawRMB.toString());
            flowBean.setInterestRMB(interestRMB.toString());
            flowBean.setBalanceRMB(balanceRMB.toString());

            context.write(NullWritable.get(),flowBean);

        }
    }


    @Override
    public int run(String[] args) throws Exception {
        Configuration conf=getConf();

        Job job=Job.getInstance(conf,this.getClass().getSimpleName());
        job.setJarByClass(March01.class);

        job.setMapperClass(March01.March01Mapper.class);
        job.setMapOutputKeyClass(Text.class);
        job.setMapOutputValueClass(NullWritable.class);

        job.setReducerClass(March01.March01Reducer.class);
        job.setOutputKeyClass(NullWritable.class);
        job.setOutputValueClass(FlowBean.class);

        job.setInputFormatClass(TextInputFormat.class);
        job.setOutputFormatClass(TextOutputFormat.class);

        TextInputFormat.addInputPath(job,new Path("F:\\test\\march01Input"));
        TextOutputFormat.setOutputPath(job,new Path("F:\\test\\march01Output"));

        return job.waitForCompletion(true)?0:1;
    }

    public static void main(String[] args) throws Exception {
        System.exit(ToolRunner.run(new March01(),args));
    }
}
